package com.sdyc.ndmp.cls.client;

import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.ReceiveTimeout;
import akka.actor.UntypedActor;
import akka.pattern.Patterns;
import akka.util.Timeout;
import com.sdyc.ndmp.cls.dtd.DataSentence;
import com.sdyc.ndmp.cls.dtd.KeywordCondition;
import com.sdyc.ndmp.cls.server.SentenceKeywordActor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * <pre>
 * Created with IntelliJ IDEA.
 * User: zachary
 * Date: 2015/11/18
 * Time: 14:25
 * </pre>
 *
 * @author zachary
 */
public class ClientActor extends UntypedActor {


    /**
     * cache
     */
    protected final Map<Class, ActorRef> ACTOR_REF_MAP = new HashMap<Class, ActorRef>(3);



    private HashMap<String, Serializable> param = null;



    protected int timeoutSeconds = 5;


    protected static Log log = LogFactory.getLog(ClientActor.class);


    public ClientActor(HashMap<String, Serializable> param) {
        this.param = param;

    }

    @Override
    public void preStart() throws Exception {
        super.preStart();

        ActorRef sentenceActorRef = getContext().actorOf(
                Props.create(SentenceKeywordActor.class, param));
        ACTOR_REF_MAP.put(DataSentence.class, sentenceActorRef);
        ACTOR_REF_MAP.put(KeywordCondition.class, sentenceActorRef);
    }

    /**
     * 客户端发送出去，以及接收服务器端返回都是走该方法。
     *
     * @param o
     * @throws Exception
     */
    @Override
    public void onReceive(Object o) throws Exception {
        if (o instanceof ReceiveTimeout) {
            log.error("接收消息超时");
            getContext().setReceiveTimeout(Duration.Undefined());
        }
        //其他的都为要发送数据
        ActorRef actorRef = ACTOR_REF_MAP.get(o.getClass());
        if(actorRef != null) {
            Future<Object> future = Patterns.ask(actorRef,
                    o,
                    new Timeout(timeoutSeconds, TimeUnit.SECONDS));
            Object result = Await.result(future,
                    Duration.create(timeoutSeconds, TimeUnit.SECONDS));

            getSender().tell(result, ActorRef.noSender());
            return;
        }

        throw new UnsupportedOperationException("message class " + o.getClass().getName() + " unsupported.");

    }

}

